package drds.binlog.prometheus.impl;

import com.google.common.base.Preconditions;
import drds.binlog.instance.core.BinlogTask;
import drds.binlog.prometheus.InstanceRegistry;
import drds.binlog.sink.EventDownStreamHandler;
import drds.binlog.sink.EventSink;
import drds.binlog.sink.entry.EntryEventSink;
import io.prometheus.client.Collector;
import io.prometheus.client.CounterMetricFamily;
import io.prometheus.client.GaugeMetricFamily;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;

import static drds.binlog.prometheus.InstanceExports.DEST_LABELS_LIST;


public class EntryCollector extends Collector implements InstanceRegistry
{

    private static final Logger logger = LoggerFactory.getLogger(SinkCollector.class);
    private static final String DELAY = "canal_instance_traffic_delay";
    private static final String TRANSACTION = "canal_instance_transactions";
    private static final String DELAY_HELP = "Traffic delay of canal instance in milliseconds";
    private static final String TRANSACTION_HELP = "Transactions counter of canal instance";
    private final ConcurrentMap<String, EntryMetricsHolder> instances = new ConcurrentHashMap<String, EntryMetricsHolder>();

    private EntryCollector()
    {
    }

    public static EntryCollector instance()
    {
        return SingletonHolder.SINGLETON;
    }

    @Override
    public List<MetricFamilySamples> collect()
    {
        List<MetricFamilySamples> mfs = new ArrayList<MetricFamilySamples>();
        GaugeMetricFamily delay = new GaugeMetricFamily(DELAY,
                DELAY_HELP, DEST_LABELS_LIST);
        CounterMetricFamily transactions = new CounterMetricFamily(TRANSACTION,
                TRANSACTION_HELP, DEST_LABELS_LIST);
        for (EntryMetricsHolder entryMetricsHolder : instances.values())
        {
            long now = System.currentTimeMillis();
            long latest = entryMetricsHolder.latestExecTime.get();
            // execTime > now，delay显示为0
            long d = (now >= latest) ? (now - latest) : 0;
            delay.addMetric(entryMetricsHolder.destLabelValues, d);
            transactions.addMetric(entryMetricsHolder.destLabelValues, entryMetricsHolder.transactionCounter.doubleValue());
        }
        mfs.add(delay);
        mfs.add(transactions);
        return mfs;
    }

    @Override
    public void register(BinlogTask binlogTask)
    {
        final String destination = binlogTask.getDestination();
        EntryMetricsHolder holder = new EntryMetricsHolder();
        holder.destLabelValues = Collections.singletonList(destination);
        EventSink sink = binlogTask.getEventSink();
        if (!(sink instanceof EntryEventSink))
        {
            throw new IllegalArgumentException("EventSink must be EntryEventSink");
        }
        EntryEventSink entrySink = (EntryEventSink) sink;
        PrometheusEventDownStreamHandler handler = assembleHandler(entrySink);
        holder.latestExecTime = handler.getLatestExecuteTime();
        holder.transactionCounter = handler.getTransactionCounter();
        Preconditions.checkNotNull(holder.latestExecTime);
        Preconditions.checkNotNull(holder.transactionCounter);
        EntryMetricsHolder old = instances.put(destination, holder);
        if (old != null)
        {
            logger.warn("Remove stale EntryCollector for binlogTask {}.", destination);
        }
    }

    @Override
    public void unregister(BinlogTask binlogTask)
    {
        final String destination = binlogTask.getDestination();
        EventSink sink = binlogTask.getEventSink();
        if (!(sink instanceof EntryEventSink))
        {
            throw new IllegalArgumentException("EventSink must be EntryEventSink");
        }
        unloadHandler((EntryEventSink) sink);
        instances.remove(destination);
    }

    private PrometheusEventDownStreamHandler assembleHandler(EntryEventSink entrySink)
    {
        PrometheusEventDownStreamHandler ph = new PrometheusEventDownStreamHandler();
        List<EventDownStreamHandler> handlers = entrySink.getHandlers();
        for (EventDownStreamHandler handler : handlers)
        {
            if (handler instanceof PrometheusEventDownStreamHandler)
            {
                throw new IllegalStateException("PrometheusEventDownStreamHandler already exists in handlers.");
            }
        }
        entrySink.addHandler(ph, 0);
        return ph;
    }

    private void unloadHandler(EntryEventSink entryEventSink)
    {
        List<EventDownStreamHandler> eventDownStreamHandlerList = entryEventSink.getHandlers();
        int i = 0;
        for (; i < eventDownStreamHandlerList.size(); i++)
        {
            if (eventDownStreamHandlerList.get(i) instanceof PrometheusEventDownStreamHandler)
            {
                break;
            }
        }
        entryEventSink.removeHandler(i);
        // Ensure no PrometheusEventDownStreamHandler
        eventDownStreamHandlerList = entryEventSink.getHandlers();
        for (EventDownStreamHandler eventDownStreamHandler : eventDownStreamHandlerList)
        {
            if (eventDownStreamHandler instanceof PrometheusEventDownStreamHandler)
            {
                throw new IllegalStateException("Multiple prometheusCanalEventDownStreamHandler exists in handlers.");
            }
        }
    }

    private static class SingletonHolder
    {
        private static final EntryCollector SINGLETON = new EntryCollector();
    }

    private class EntryMetricsHolder
    {
        private AtomicLong latestExecTime;
        private AtomicLong transactionCounter;
        private List<String> destLabelValues;
    }

}
